/**
 * 
 */
package xiaojian.toolkit.transport;

import java.util.concurrent.BlockingQueue;

import javax.management.InstanceNotFoundException;
import javax.management.MBeanRegistrationException;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;

import org.apache.commons.collections4.Closure;
import org.apache.mina.core.session.IoSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jmx.export.MBeanExporter;

/**
 * @author hp
 *
 */
public abstract class AbstractEndpoint implements MutableEndpoint {
	
	private static final Logger logger = 
    	LoggerFactory.getLogger(AbstractEndpoint.class);

	protected	Closure			messageClosure;
	protected	Receiver		nextReceiver;
	protected 	IoSession		session = null;
    
    protected 	BlockingQueue<Object>	sharedPendings = null;
    
    protected 	BlockingQueue<Object>	pendings = null;
    
    protected  	MBeanExporter	mbeanExport = null;
    
    protected	String	objectNamePrefix;
    
    protected	ObjectName	objectName = null;
    
	/**
	 * @return the objectNamePrefix
	 */
	public String getObjectNamePrefix() {
		return objectNamePrefix;
	}

	/**
	 * @param objectNamePrefix the objectNamePrefix to set
	 */
	public void setObjectNamePrefix(String objectNamePrefix) {
		this.objectNamePrefix = objectNamePrefix;
	}

	/**
	 * @return the mbeanExport
	 */
	public MBeanExporter getMbeanExport() {
		return mbeanExport;
	}

	/**
	 * @param mbeanExport the mbeanExport to set
	 */
	public void setMbeanExport(MBeanExporter mbeanExport) {
		this.mbeanExport = mbeanExport;
	}

	/**
	 * @see com.skymobi.transport.Sender#send(java.lang.Object)
	 */
	public void send(Object bean) {
    	pendings.add(bean);
	}

	/**
	 * @see com.skymobi.transport.Receiver#messageReceived(org.apache.mina.core.session.IoSession, java.lang.Object)
	 */
	public void messageReceived(IoSession session, Object msg) {
		if ( null != messageClosure) {
			this.messageClosure.execute( TransportUtils.attachSender(msg, this) );
		}
		if ( null != nextReceiver ) {
			this.nextReceiver.messageReceived(session, msg);
		}
	}

	public void setIoSession(IoSession session) {
		this.session = session;
		if ( null != objectName && null != mbeanExport ) {
			try {
				mbeanExport.getServer().unregisterMBean(objectName);
			} catch (MBeanRegistrationException e) {
				logger.error("setIoSession:", e);
				e.printStackTrace();
			} catch (InstanceNotFoundException e) {
				logger.error("setIoSession:", e);
			}
		}
		if ( null != session && null != mbeanExport ) {
			try {
				objectName = new ObjectName(objectNamePrefix + ",session="+
						session
						.toString().replaceAll(":", "/")
						.replaceAll(",", "")
						.replaceAll("=>", "-")
						);
				mbeanExport.registerManagedResource(this, objectName);
			} catch (MalformedObjectNameException e) {
				logger.error("setIoSession:", e);
			} catch (NullPointerException e) {
				logger.error("setIoSession:", e);
			}
		}
	}

	/* (non-Javadoc)
	 * @see com.skymobi.transport.Endpoint#stop()
	 */
	public void stop() {
		if ( null != objectName && null != mbeanExport ) {
			try {
				mbeanExport.getServer().unregisterMBean(objectName);
			} catch (MBeanRegistrationException e) {
				logger.error("setIoSession:", e);
				e.printStackTrace();
			} catch (InstanceNotFoundException e) {
				logger.error("setIoSession:", e);
			}
		}
	}

	public void setMessageClosure(Closure closure) {
		this.messageClosure = closure;
	}

	public void setSendQueue(BlockingQueue<Object> queue) {
		this.pendings = queue;
	}

	public void setSharedSendQueue(BlockingQueue<Object> queue) {
		this.sharedPendings = queue;
	}
	
	/**
	 * @param nextReceiver the nextReceiver to set
	 */
	public void setNextReceiver(Receiver nextReceiver) {
		this.nextReceiver = nextReceiver;
	}

	public int getPendingCount() {
		if ( null != this.pendings ) {
			return	this.pendings.size();
		}
		else {
			return	-1;
		}
	}
}
